package org.apache.rocketmq.example.kang.consumer.sample;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragelyByCircle;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class PushConsumerTest {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(SampleConfig.CONSUMER_GROUP);
        consumer.setNamesrvAddr(SampleConfig.NAMESRV_ADDR);
        // 指定初次启动时从什么地方消费，默认是最新的消息开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(SampleConfig.TOPIC, "*");
        // 指定队列负载机制，默认平均分配
        consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragelyByCircle());
       /*
        这里设置 consumeMessageBatchMaxSize 的值大于 pullBatchSize 的主要目的就是验证每一次拉取的消息
        因为如果 consumeMessageBatchMaxSize 大于 pullBatchSize，那每次批处理的消息条数等于 pullBatchSize，
        如果 consumeMessageBatchMaxSize 小于 pullBatchSize，会在客户端分页，然后尽最大可能一次传入 consumeMessageBatchMaxSize 条消息
        */
        consumer.setPullBatchSize(2); // 消息客户端一次向 Broker 发送拉取消息每批返回最大的消息条数，默认为 32
        consumer.setConsumeMessageBatchMaxSize(4); // 提交到消息消费监听器中的消息条数，默认为 1。

        // 设置消息监听器，即消息处理逻辑，最终返回 CONSUME_SUCCESS（成功消费）或 RECONSUME_LATER（需要重试）
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    System.out.printf("本次消费消息条数为:%s %n",msgs.size());
                    System.out.printf("%s Receive New Messages: %s %n",
                            Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Throwable e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
